package com.flink.unit;

import com.flink.KafkaSinkES;
import com.flink.conn.SinkLogToES;
import com.flink.entity.LogInfoDTO;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.format.DateTimeFormatter;
import java.time.temporal.TemporalAccessor;
import java.util.*;

/**
 * @Author LT-0024
 * @Date 2020/8/18 11:15
 * @Version 1.0
 */
public class LogInfoStreamUnit {

    /**
     * 使用指定类初始化日志对象
     */
    private static Logger logger = LoggerFactory.getLogger(KafkaSinkES.class);
    public static final String STRING = "0";

    public static void startStream(StreamExecutionEnvironment env, Constant constant, final String topic, final String table, String group, Properties props) {
        props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
        FlinkKafkaConsumer011<String> stream = new FlinkKafkaConsumer011<String>(
                topic,
                new SimpleStringSchema(),
                props);
        if (!STRING.equals(constant.startFromTimestamp)) {
            stream.setStartFromTimestamp(Long.parseLong(constant.startFromTimestamp));
        }
        // 从最后一次开始消费
        stream.setStartFromGroupOffsets();
        // 获取数据源
        DataStreamSource<String> stringDataStreamSource = env.addSource(stream);
        // 打印
        stringDataStreamSource.print();
        // 先过滤掉 异常 信息 at开头的寨信息
        SingleOutputStreamOperator<String> filterLine = stringDataStreamSource.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String line) throws Exception {
                String[] split = line.split("\\$\\$");
                // 不符合log日志格式过滤掉
                if (split.length != 11) {
                    return false;
                }
                // 过滤掉没有traceID的数据
                if (StringUtils.isBlank(split[7])) {
                    return false;
                }
                return true;
            }
        });
        //过滤数据
        SingleOutputStreamOperator<LogInfoDTO> empStream = filterLine
                // 并行10
                .setParallelism(Integer.parseInt(constant.parallelism))
                // 解析过程
                .map(new MapFunction<String, LogInfoDTO>() {
                    //解析字符串成JSON对象
                    @Override
                    public LogInfoDTO map(String string) throws Exception {
                        LogInfoDTO logInfoBean = new LogInfoDTO();
                        logInfoBean.setLogID(UUID.randomUUID().toString());
                        String[] split = string.split("\\$\\$");
                        logger.info(split.length + "-----" + string);
                        for (String str : split) {
                            try {
                                // 匹配时间
                                boolean logDateFlag = str.matches("\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}.\\d{3}");
                                if (logDateFlag) {
                                    String logDate = split[0];
                                    TemporalAccessor parse = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS").parse(logDate);
                                    String dateString = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS").format(parse);
                                    logInfoBean.setLogDate(Long.valueOf(dateString));
                                }
                                // 匹配集团号 都是数字
                                boolean groupIDFlag = str.matches("^[0-9]*[0-9][0-9]*$");
                                if (groupIDFlag) {
                                    logInfoBean.setGroupID(str);
                                }
                                // 提取单号
                                String[] billNos = split[split.length - 1].split("billNo=");
                                if (billNos.length > 1) {
                                    String[] split1 = billNos[1].split(",");
                                    logInfoBean.setBillNo(split1[0]);
                                }
                                // 提取库存单号
                                String[] vouchers = split[split.length - 1].split("voucherNo=");
                                if (vouchers.length > 1) {
                                    String[] split1 = vouchers[1].split(",");
                                    logInfoBean.setVoucherNo(split1[0]);
                                }
                            } catch (Exception e) {
                                logger.error(e.getMessage());
                            }
                            logInfoBean.setUuid(split[7]);
                            logInfoBean.setServiceName(split[3]);
                            logInfoBean.setFunctionName(split[split.length - 2]);
                            logInfoBean.setInfo(split[split.length - 1]);
                        }
                        return logInfoBean;
                    }
                });

        // 开个一分钟的窗口去聚合
        logger.info("开启timeWindow" + table + topic);
        empStream.timeWindowAll(Time.seconds(Long.parseLong(constant.timeWindow))).apply(new AllWindowFunction<LogInfoDTO, List<LogInfoDTO>, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<LogInfoDTO> values, Collector<List<LogInfoDTO>> out) throws Exception {
                List<LogInfoDTO> orderDetails = new ArrayList();
                values.forEach(x -> {
                    orderDetails.add(x);
                });
                Set set = new HashSet();
                List<LogInfoDTO> newList = new ArrayList<>();
                set.addAll(orderDetails);
                newList.addAll(set);
                if (orderDetails.size() > 0) {
                    logger.info("10 秒内收集到" + topic + " 的数据条数是：" + orderDetails.size());
                    out.collect(newList);
                }
            }
        }).addSink(new SinkLogToES(table, topic));
        empStream.print(); //调度输出
    }
}
